package com.ziroom.crm.mq.test;

import com.rabbitmq.client.Channel;
import com.ziroom.crm.mq.rabbit.spring.core.ReliableMessageListenerContainer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Random;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.Consumer;

@Service
@SuppressWarnings("ALL")
@Slf4j
public class ConsumerDemo implements InitializingBean, DisposableBean {

    ThreadLocalRandom random = ThreadLocalRandom.current();
    Set<String> successSet = new CopyOnWriteArraySet<>();
    Set<String> failSet = new CopyOnWriteArraySet<>();
    Set<String> set = new CopyOnWriteArraySet<>();

    @Autowired
    private ConnectionFactory connectionFactory;

    private ReliableMessageListenerContainer messageListenerContainer;

    @Override
    public void afterPropertiesSet() throws Exception {
        Queue queue = new Queue(Pkg.queue);
        messageListenerContainer = new ReliableMessageListenerContainer();
        messageListenerContainer.setConnectionFactory(connectionFactory);
        messageListenerContainer.setQueues(queue);
        messageListenerContainer.setConcurrency("5");
        messageListenerContainer.setPrefetchCount(100);
        messageListenerContainer.setMessageListener((Message message) -> {
            consume(message);
        });
        messageListenerContainer.setCountCallback(new Consumer<Integer>() {
            @Override
            public void accept(Integer count) {
                log.info("当前count={}", count);
            }
        });
        // messageListenerContainer.setCallback(new Consumer<List<? extends com.ziroom.crm.mq.core.data.Message>>() {
        //     @Override
        //     public void accept(List<? extends com.ziroom.crm.mq.core.data.Message> messages) {
        //         if (!CollectionUtils.isEmpty(messages)) {
        //             for (com.ziroom.crm.mq.core.data.Message message : messages) {
        //                 if (message instanceof RabbitMessage) {
        //                     System.out.println("接受到Rabbit消息 = " + JSON.toJSONString(message));
        //                 } else {
        //                     System.out.println("接受到消息 = " + JSON.toJSONString(message));
        //                 }
        //             }
        //         }
        //     }
        // });
        messageListenerContainer.start();
        // simpleMessageListenerContainer1.stop();
    }


    private ChannelAwareMessageListener channelAwareMessageListener() {
        return (Message message, Channel channel) -> {
            consume(message);
        };
    }


    private void consume(Message message) {
        String messageId = message.getMessageProperties().getMessageId();
        set.add(messageId);
        String s = new String(message.getBody());
        int i = new Random().nextInt(10);
        if ((s.contains("fail") || s.contains("Fail")) && i < 5) {
            failSet.add(messageId);
            System.out.println("consumer msg:fail " + s + successSet.size() + "/" + failSet.size() + "/" + set.size());
            throw new RuntimeException("test mq fail:" + failSet.size());
        }
        // 消费成功
        // 重试结束后，successSet.size + 死信数 = 发送端成功数
        // 消费结束，重试前，successSet.size + backup数 = 发送端成功数
        successSet.add(messageId);
        failSet.remove(messageId);
        System.out.println("consumer msg: success" + s + successSet.size() + "/" + failSet.size() + "/" + set.size());
    }


    @Override
    public void destroy() throws Exception {
        if (messageListenerContainer.isRunning()) {
            messageListenerContainer.stop();
            log.info("====停止: isRunning={}", messageListenerContainer.isRunning());
        }
    }

}